﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Threading;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.Diagnostics;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure.StorageClient;
using Microsoft.WindowsAzure.Search;

namespace Master
{
    public class WorkerRole : RoleEntryPoint
    {
        private int current;

        private const String PARSER = "parser";
        private const String MASTER = "master";
        private const String INVERTER = "inverter";
        private const String START = "start";
        private const String DONE = "done";

        // number of documents for one partition
        private int partition;
        // number of ranges for a term
        private int range;

        private TotalinfoContext context;

        private Stopwatch watch;

        public override void Run()
        {
            // account
            CloudStorageAccount dataAccount = CloudStorageAccount.FromConfigurationSetting("DataConnectionString");
            CloudStorageAccount indexAccount = CloudStorageAccount.FromConfigurationSetting("IndexConnectionString");

            // client
            CloudBlobClient dataClient = dataAccount.CreateCloudBlobClient();
            CloudBlobClient indexClient = indexAccount.CreateCloudBlobClient();
            CloudQueueClient queueClient = indexAccount.CreateCloudQueueClient();

            watch = new Stopwatch();
 
            while (true)
            {
                CloudQueue queue = queueClient.GetQueueReference(MASTER);
                queue.CreateIfNotExist();

                CloudQueueMessage m = queue.GetMessage();
                if (m != null && m.AsString.Contains(START))
                {
                    String[] info = Tokenizer.CommaTokenize(m.AsString);
                    partition = Convert.ToInt32(info[1]);
                    range = Convert.ToInt32(info[2]);

                    watch.Start();
                    queue.DeleteMessage(m);

                    current = 0;

                    int blobNum = 0;
                    IEnumerable<CloudBlobContainer> containers = dataClient.ListContainers();
                    Debug.WriteLine("container Number: " + containers.ToList().Count);
                    foreach (CloudBlobContainer container in containers)
                    {
                        blobNum += container.ListBlobs().ToList().Count;
                        //if (blobNum >= 1000)
                        //    break;
                    }
                    Debug.WriteLine("blob Number: " + blobNum);


                    int num = (blobNum - 1) / partition + 1;

                    // control parser
                    CloudQueue parserQueue = queueClient.GetQueueReference(PARSER);
                    parserQueue.CreateIfNotExist();

                    for (int i = 0; i < num - 1; i++)
                    {
                        String message = range + "," + i * partition + "," + ((i + 1) * partition - 1);
                        CloudQueueMessage msg = new CloudQueueMessage(message);
                        parserQueue.AddMessage(msg);
                    }

                    Debug.WriteLine("Assign Parser");

                    // control inverter
                    CloudQueue inverterQueue = queueClient.GetQueueReference(INVERTER);
                    for (int i = 0; i < range; i++)
                    {
                        CloudQueueMessage rangeMsg = new CloudQueueMessage(i + "," + num);
                        inverterQueue.AddMessage(rangeMsg);
                    }
                    
                    // last message
                    String lastMessage = range + "," + (num - 1) * partition + "," + (blobNum - 1);
                    CloudQueueMessage lastmsg = new CloudQueueMessage(lastMessage);
                    parserQueue.AddMessage(lastmsg);

                    Debug.WriteLine("Assign Inverter");

                    context.Add(blobNum, range);
                }

                if (m != null && m.AsString.Contains(DONE))
                {
                    Debug.WriteLine("Inverter " + Tokenizer.CommaTokenize(m.AsString)[1] + " Done");

                    queue.DeleteMessage(m);
                    current++;
                    if (current == range)
                        Finish();
                }

                Thread.Sleep(2000);
            }
        }

        private void Finish()
        {
            watch.Stop();
            Debug.WriteLine("*************Index Construction Completely*************");
            Debug.WriteLine("Time: " + watch.ElapsedMilliseconds + "ms");

            CloudStorageAccount indexAccount = CloudStorageAccount.FromConfigurationSetting("IndexConnectionString");
            CloudBlobClient client = indexAccount.CreateCloudBlobClient();
            CloudBlobContainer container = client.GetContainerReference("log");
            container.CreateIfNotExist();
            CloudBlob blob = container.GetBlobReference("index");
            blob.UploadText("Index Complete");
        }

        public override bool OnStart()
        {
            // Set the maximum number of concurrent connections 
            ServicePointManager.DefaultConnectionLimit = 12;

            // For information on handling configuration changes
            // see the MSDN topic at http://go.microsoft.com/fwlink/?LinkId=166357.

            #region Setup CloudStorageAccount Configuration Setting Publisher

            CloudStorageAccount.SetConfigurationSettingPublisher(
                (configName, configSetter) =>
                {
                    // Provide the configSetter with the initial value
                    configSetter(RoleEnvironment.GetConfigurationSettingValue(configName));
                    RoleEnvironment.Changed += (s, arg) =>
                    {
                        if (arg.Changes.OfType<RoleEnvironmentConfigurationSettingChange>()
                            .Any((change) => (change.ConfigurationSettingName == configName)))
                        {
                            if (!configSetter(RoleEnvironment.GetConfigurationSettingValue(
                                configName)))
                                // In this case, the change to the storage account credentials
                                // in the service configuration is significant enough that the
                                // role needs to be recycled in order to use the latest settings.
                                // (for example, the endpoint has changed)
                                RoleEnvironment.RequestRecycle();
                        }
                    };
                });
            #endregion

            CloudStorageAccount indexAccount = CloudStorageAccount.FromConfigurationSetting("IndexConnectionString");
            CloudTableClient tableClient = indexAccount.CreateCloudTableClient();
            tableClient.CreateTableIfNotExist("docinfo");
            context = new TotalinfoContext(indexAccount.TableEndpoint.ToString(), indexAccount.Credentials);

            return base.OnStart();
        }
    }
}
